package com.alison.sparkstream.source

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

object E2_stream_file {

  """
    |文件流是指从文件系统中实时读取数据，并将数据流转换为 DStream 的一种输入源。这对于监控目录中不断产生新数据的场景非常有用。
    |在 Spark Streaming 中，可以通过 StreamingContext.textFileStream(directory) 来创建一个文件流。该方法会监控指定目录下的新文件，并读取其中的数据。
    |
    |程序启动后，它会开始监控指定目录，并读取在监控目录中新产生文件的数据。我们可以测试一下，进入设定的监控目录，创建一个文本文件：
    |cd /Volumes/BOOTCAMP/test
    |vi text.txt
    |
    |hello java
    |hello spark
    |hello flink
    |""".stripMargin

  def main(args: Array[String]): Unit = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("ncExample")
    // 初始化 StreamingContext，设置微批处理的时间间隔为 3 秒
    val streamingContext = new StreamingContext(sparkConf, Seconds(3))
    // 监控的目录路径
    var directory = "/Volumes/BOOTCAMP/test"
    directory = this.getClass.getResource("/").toString + "testfile/"
    directory = directory.replace("file:", "file://")
    directory = "D:\\workspace\\lab\\learnbigdata\\learnspark\\sparkstream\\src\\main\\resources\\testfile"
    println(directory)
    // 创建文件流，从指定目录读取数据
    val lineStreams = streamingContext.textFileStream(directory)
    // 对 DStream 进行转换操作，将每一行数据做切分，形成一个个单词
    val wordStreams = lineStreams.flatMap(_.split(" "))
    // 将单词映射成元组（word,1）
    val wordAndOneStreams = wordStreams.map((_, 1))
    // 将相同的单词次数做统计
    val wordAndCountStreams = wordAndOneStreams.reduceByKey(_ + _)
    // 打印结果
    wordAndCountStreams.print()
    // 启动 StreamingContext
    streamingContext.start()
    // 等待应用程序终止（遇错停止，否则不断工作）
    streamingContext.awaitTermination()
  }
}
